package service

import (
	"context"
	"encoding/json"
	"task/driver"
	"time"

	"task/models"

	"github.com/apache/pulsar-client-go/pulsar"
	log "github.com/sirupsen/logrus"
)

const (
	channelLen = 100000
	workerNum  = 5
)

var (
	// ExecuteEventCh execute event channel.
	// 执行事件通道
	ExecuteEventCh = make(chan pulsar.ConsumerMessage, channelLen)
	// ProcessorMap processor map.
	ProcessorMap = make(map[int]*models.Processor)
)

// ListenExecuteEvent listen execute event.
func ListenExecuteEvent(consumer pulsar.Consumer) {
	for i := 0; i < workerNum; i++ {
		go func() {
			for {
				select {
				case cm := <-ExecuteEventCh:
					// 记录开始时间
					start := time.Now()
					msg := cm.Message
					// 消息第一时间确认 避免代码提前返回 导致消息重复消费
					consumer.Ack(msg)
					log.Infof("Received execute message: %s channel len: %d", string(msg.Payload()), len(ExecuteEventCh))
					// 解析执行消息
					event := new(models.ProcessEvent)
					if err := json.Unmarshal(msg.Payload(), &event); err != nil {
						log.Errorf("Unmarshal message error: %v", err)
						continue
					}
					// 处理执行事件
					if err := processExecuteEvent(context.Background(), event); err != nil {
						// 处理错误处理
						handleProcessError(event, err)
					}
					// metrics监控记录
					recordMetrics(float64(len(ExecuteEventCh)), start, event)
				}
			}
		}()
	}
}

// SetProcessorMap set processor map.
func SetProcessorMap() {
	log.Infof("Set processor map...")
	var (
		processors []*models.Processor
		err        error
	)
	if processors, err = driver.GetAllProcessors(); err != nil {
		log.Panic("SyncProcessorMap error: %v", err)
	}
	for _, processor := range processors {
		ProcessorMap[processor.Id] = processor
	}
}

// SyncProcessorMap sync processor map.
func SyncProcessorMap() {
	for {
		SetProcessorMap()
		time.Sleep(5 * time.Minute)
	}
}
